package com.youshang520i.consuldiscoveryconsumer.listener;

import com.youshang520i.consuldiscoveryconsumer.service.RocketmqService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * cxt 相关业务
 *
 * @author
 */
@Component
public class LearningMQListener {
    private final Logger logger = LoggerFactory.getLogger(LearningMQListener.class);
    @Value("${rocketmq.name-server}")
    private String nameSrvAddr;
    @Autowired
    private RocketmqService mqService;

    @PostConstruct
    public void init() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-youshang-consumer");
        consumer.setNamesrvAddr(nameSrvAddr);
        consumer.subscribe(MQTopicConstants.TOPIC_YOUSHANG, "*");
        consumer.setConsumeMessageBatchMaxSize(1);
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                boolean saveDb = msg.getReconsumeTimes() >= 4;
                try {
                    processMsg(msg);
                    logger.info("【消息处理成功】消息id：{}， 消息主题：{}，消息标签：{}， 重试次数：{}， 消息内容：{}", msg.getMsgId(), msg.getTopic(), msg.getTags(), msg.getReconsumeTimes(), new String(msg.getBody()));
                } catch (Exception e) {
                    logger.error("【消息处理失败】消息id：{}， 消息主题：{}，消息标签：{}， 重试次数：{}， 消息内容：{}", msg.getMsgId(), msg.getTopic(), msg.getTags(), msg.getReconsumeTimes(), new String(msg.getBody()));
                    logger.error(e.getMessage(), e);

                    if (saveDb){
                        logger.info("【消息错误次数过多，记录等待重发】消息id：{}， 消息主题：{}，消息标签：{}， 重试次数：{}， 消息内容：{}", msg.getMsgId(),
                                msg.getTopic(), msg.getTags(), msg.getReconsumeTimes(), new String(msg.getBody()));

                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        logger.info(">>>>>>>>>>>> start rocketmq listener");
    }

    private void processMsg(MessageExt msg){
        switch (msg.getTags()){
            case MQTagConstants.TAG_YOUSHANG_LEARNING: {
                logger.info("开始处理test-yousahng-consume任务: {}, url: {}", "test更新同步商品数据", MQTagConstants.TAG_YOUSHANG_LEARNING);
                String keys = msg.getKeys();
                mqService.receive(new String(msg.getBody()));
                break;
            }
            default:
                throw new RuntimeException("未定义处理消息");
        }
    }
}
